﻿//========================================================= 
/**@file rocketmq_extractor.h
 * @brief RocketMq常用功能封装
 *
 * - 支持的功能:
 *  - 配合lmdb做消息去重过滤
 *  - 提供顺序消息的收发机制
 * .
 * 
 * @date 2018-08-30   09:49:15
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_ROCKETMQ_EXTRACTOR_H_
#define _LIBZHOUYB_ROCKETMQ_EXTRACTOR_H_
//--------------------------------------------------------- 
#include <DefaultMQPushConsumer.h>
#include <DefaultMQProducer.h>
#include <MQMessageListener.h>
using namespace rocketmq;

#include "../lmdb/lmdb_extractor.h"
using namespace zhou_yb::wrapper::lmdb_extractor;
//--------------------------------------------------------- 
namespace zhou_yb {
namespace wrapper {
namespace rocketmq_extractor {
//--------------------------------------------------------- 
/// 默认的顺序消息选择器
class OrderlyMsgQueueSelecter : public MessageQueueSelector
{
public:
    MQMessageQueue select(const std::vector<MQMessageQueue>& mqs, const MQMessage& msg, void* arg)
    {
        size_t orderId = *(static_cast<size_t*>(arg));
        size_t index = orderId % mqs.size();
        return mqs[index];
    }
};
//--------------------------------------------------------- 
/**
 * @brief RocketMq辅助函数
 * @date 2018-08-30 11:10
 */
class RocketMqHelper
{
public:
    static OrderlyMsgQueueSelecter OrderlySelector;

    static ByteArray LAST_OFFSET;
    static ByteArray LAST_AND_BOOT_FIRST_OFFSET;
    static ByteArray MIN_OFFSET;
    static ByteArray MAX_OFFSET;
    static ByteArray FIRST_OFFSET;
    static ByteArray TIMESTAMP;

    static size_t COMSUMER_FROM_TABLE_SIZE;
    static ByteArray COMSUMER_FROM_TABLE[6];
public:
    static ConsumeFromWhere FromWhere(const ByteArray& fromWhere)
    {
        for(size_t i = 0; i < COMSUMER_FROM_TABLE_SIZE; ++i)
        {
            if(StringConvert::Compare(fromWhere, COMSUMER_FROM_TABLE[i], true))
                return static_cast<ConsumeFromWhere>(i);
        }
        return CONSUME_FROM_LAST_OFFSET;
    }
    static SendResult Send(DefaultMQProducer& producer, MQMessage& msg, size_t orderId)
    {
        return producer.send(msg, &OrderlySelector, static_cast<void*>(&orderId));
    }
    static SendResult Send(DefaultMQProducer& producer, MQMessage& msg, size_t orderId,
        int autoRetryTimes, bool bActive)
    {
        return producer.send(msg, &OrderlySelector, static_cast<void*>(&orderId), autoRetryTimes, bActive);
    }
    static bool IsRepeated(lmdb_string_iterator& itr, const string& msgId, bool& isRepeated)
    {
        if(itr.is_valid())
        {
            if(itr.get(msgId))
            {
                isRepeated = true;
                return true;
            }
            if(itr.set(msgId, ""))
            {
                isRepeated = false;
                return true;
            }
        }
        return false;
    }
};
//--------------------------------------------------------- 
} // namespace rocketmq_extractor 
} // namespace wrapper
} // namespace zhou_yb 
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_ROCKETMQ_EXTRACTOR_H_
//========================================================= 